package com.gosuncn.FileCopy;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.net.URLEncoder;
import java.util.*;
import java.util.concurrent.*;

/***
 *HttpURLConnection +多线程 进行文件传输
 * @author zxl
 * @date 2021/9/1
 * @param
 * @return
*/
public class CallableTest {


    //连接超时时间
    private static int HttpTimeOut = 6000;
    //请求方式
    private static String Method_Post ="POST";
    private ArrayBlockingQueue<Future<Map<String, FileInputStream>>> queue;
    private String serverUrl = "http://localhost:8081/file/upload2";


    /**
     * 通过http请求建立连接
     * @param serverUrl
     * @param fileName
     */
    public HttpURLConnection initConnection(String serverUrl, String fileName){
        HttpURLConnection conn =null;
        try {
            //得到网络访问对象HttpURLConnection
            conn = (HttpURLConnection) new URL(serverUrl).openConnection();
            //从主机读取数据的超时时间
            conn.setReadTimeout(HttpTimeOut);
            // 连接主机的超时时间
            conn.setConnectTimeout(HttpTimeOut);
            //请求方式 post   这里如果用get请求会有异常
            conn.setRequestMethod(Method_Post);
            //默认是ture  可以调用getInputStream（）方法
            conn.setDoInput(true);
            //默认是false 设置ture 可以调用getOutputStream（）方法
            conn.setDoOutput(true);
            //分段传输，不然默认本地文件全部写入再进行传输， 不然会报OOM
            conn.setChunkedStreamingMode(1024);
            conn.setRequestProperty("Content-Type","application/x-www-form-urlencoded;charset=UTF-8");
            //URLEncoder.encode(fileName,"UTF-8")防止中文乱码
            conn.setRequestProperty("fileName", URLEncoder.encode(fileName,"UTF-8"));
            //建立连接
            conn.connect();
            return conn;
        } catch (IOException e) {
            System.out.println("服务器连接失败");
            e.printStackTrace();
            return null;
        }
    }



    //读文件
    public class MyCallableProducer implements Callable<Map<String, FileInputStream>>{
        private File file;
        private FileInputStream fis =null;
        private Map<String,FileInputStream> fileMap = new HashMap<String, FileInputStream>();

        public MyCallableProducer(File file){
            this.file = file;
        }

        @Override
        public Map<String, FileInputStream> call() throws Exception {
            System.out.println(Thread.currentThread().getName()+"线程开始读文件:"+file.getName());
            fis = new FileInputStream(file);
            //获取文件的路径
            String filePath = file.getAbsolutePath();
            // D:\oldpath\zxl\123.txt 获取给定目录下的所有文件目录 zxl\123.txt
            String folder = filePath.substring(filePath.indexOf("\\", 3)+1);
            System.out.println(folder);
            //存入map中
            fileMap.put(folder,fis);
            System.out.println(Thread.currentThread().getName()+"线程读文件:"+file.getName()+"结束");
            return fileMap;
        }
    }


     //写文件
     public class MyCallableConsumer implements Runnable{

        private CountDownLatch latch ;
        private BlockingQueue<Future<Map<String,FileInputStream>>> queue;
        private FileInputStream fis = null;
        private String fileName = "";
        private String serverUrl="";
        private HttpURLConnection conn = null;
        private OutputStream os = null;


        public  MyCallableConsumer(BlockingQueue<Future<Map<String,FileInputStream>>> queue,String serverUrl,CountDownLatch latch){
            this.queue = queue;
            this.serverUrl = serverUrl;
            this.latch = latch;
        }


        @Override
        public void run() {
            try {
                //阻塞队列中移除一个任务存放在Future
                Future<Map<String, FileInputStream>> future = queue.take();
                //future.get()只有所有任务都完成后，才能使用get按照任务的提交顺序得到返回结果，
                //但是在这这里，因为我先是把任务都存在阻塞队列，一个任务存一次，就来到这里取出来，这时的future所有的任务只是单独的一个，故不需要等待
                Map<String, FileInputStream> map = future.get();
                //方法返回值是Map中key值的集合
                Set<String> set = map.keySet();
                //获取一个迭代器中当前位置的对象  并迭代器中后面是否还有东西
                for (Iterator<String> iter = set.iterator(); iter.hasNext();){
                    //第一次调用Iterator 的next()方法时，它返回序列的第一个元素  。 指针会自动向后移动
                    fileName = iter.next();
                    //获取文件流
                    fis = map.get(fileName);
                    System.out.println(Thread.currentThread().getName()+"线程正在写文件"+fileName);

                    //通过url访问与服务器建立连接
                    conn = initConnection(serverUrl, fileName);
                    if (conn!=null){
                        os = conn.getOutputStream();
                        byte[] buffer = new byte[1024];
                        int len;
                        while ((len = fis.read(buffer)) !=-1){
                            os.write(buffer,0,len);
                        }
                        System.out.println("响应结果: "+conn.getResponseCode());
                    }
                    System.out.println(Thread.currentThread().getName()+"线程写文件"+fileName+"完成");
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //执行减一
                latch.countDown();
                if (os !=null){
                    try {
                        os.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if (fis !=null){
                    try {
                        fis.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if (conn !=null){
                    conn.disconnect();
                }
            }
        }

    }



    /**
     * 给服务器发送文件流
     * @param filePathList
     */
    public void  sendHttpFile(List<File> filePathList,int filenumber) throws InterruptedException {
        //计数器
        CountDownLatch latch = new CountDownLatch(filenumber);
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                4, //线程池大小  根据你的cpu核数最大
                24, //最大核心线程 根据你当前cpu核数确定
                3, //超时等待
                TimeUnit.SECONDS, //单位
                new LinkedBlockingDeque<>(Integer.MAX_VALUE), //阻塞队列
                Executors.defaultThreadFactory()/*, //线程工厂
                new ThreadPoolExecutor.DiscardPolicy()*/); //队列满了，尝试与第一个线程竞争，不会抛出异常

        //创建阻塞队列为100大小
        queue = new ArrayBlockingQueue<>(100);
        //开始时间
        long start = System.currentTimeMillis();
        for (int i = 0; i<filePathList.size(); i++){
            File file = filePathList.get(i);
            //如果是文件
            if (file.isFile()){
                //提交给了Future，Future异步替我完成这个读文件任务。
                // 如果在这里future.get()这时就必须要等待所有 的任务执行完才按顺序输出
                Future<Map<String, FileInputStream>> future = pool.submit(new MyCallableProducer(file));
                //添加进入阻塞队列
                queue.add(future);
                //写入响应流
//                CompletableFuture.runAsync(new MyCallableConsumer(queue,serverUrl,latch),pool);
                //执行写入操作
                pool.execute(new MyCallableConsumer(queue,serverUrl,latch));

            }
        }
        try {
            //计数器归零后再执行以下任务，不然在多并发下，有些线程提前完成后就关闭连接池，后续的线程就无法执行
            latch.await();
            //结束时间
            long end = System.currentTimeMillis();
            System.out.println("-------------文件同步总共所用时间："+(end-start));
            System.out.println("-----------所有的文件都已经同步成功-------------");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            //关闭线程池
            pool.shutdown();
        }

    }





    public static void main(String[] args) throws Exception {
//        File file = new File("D:\\dirA");
//        System.out.println(file.listFiles());
//        File[] files = file.listFiles();
//        for (File file1: files){
//            String absolutePath = file1.getAbsolutePath();
//
//            System.out.println(absolutePath.substring(absolutePath.indexOf("\\", 3)));
//            System.out.println(file1.getPath());
//        }
//        File[] file1 = file.listFiles();
//        for (int i = 0; i < file1.; i++) {
//            System.out.println(file1[i]);
//        }

        String folder = "D:\\dirA";
        CallableTest callableTest = new CallableTest();
        File file1 = new File(folder);
        FileUtils files = new FileUtils(file1);
        //获取目录下所有的文件
        ArrayList<File> allFileByFolder = files.getAllFile();
        //获取目录下的所有文件总数
        int filenumber = allFileByFolder.size();
        //执行读写操作
        callableTest.sendHttpFile(allFileByFolder,filenumber);

    }


}
